package fragment;/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.table.api.TableEnvironment;
import utils.PasswordUtils;


/**
 * <pre>
 * --ORACLE_HOST 10.1.10.135 --ORACLE_PORT 1521 --ORACLE_USER DAAS_ODS --ORACLE_PASSWORD YlYkalBDZngwTVVpbmc= --ORACLE_DATABASENAME ods --ORACLE_SCHEMANAME RTDM --ORACLE_TABLESNAME RTDM.TA_WX_RT_MULT_KX31_20214
 * </pre>
 */
public class OracleSource implements SqlFragment {

    @Override
    public String createConnector(ParameterTool pt, TableEnvironment tableEnv) {
        String oracleHost = pt.get("ORACLE_HOST");
        String oraclePort = pt.get("ORACLE_PORT");
        String oracleUser = pt.get("ORACLE_USER");
        String oraclePassword = PasswordUtils.decode(pt.get("ORACLE_PASSWORD"));
        String oracleDatabasename = pt.get("ORACLE_DATABASENAME");
        String oracleSchemaname = pt.get("ORACLE_SCHEMANAME");
        String oracleTablesname = pt.get("ORACLE_TABLESNAME");
        String startMode = pt.get("STARTMODE", "initial");

        StringBuilder optionalParameters = new StringBuilder();
        if (startMode.equalsIgnoreCase("specific-offset")) {
            String startupOffset = pt.get("SPECIFIC_OFFSET");
            optionalParameters.append(
                    String.format("'scan.startup.specific-offset' = '%s',\n", startupOffset));
        }

        String sourceSql = String.format("CREATE TABLE `table_4827_oracle`(\n"
                        + "    `type` int,"
                        + "    `phase` int,"
                        + "    `data_source` string,"
                        + "    `db` string,"
                        + "    `schema` string,"
                        + "    `table` string,"
                        + "    `record` bytes,"
                        + "    `ddl` bytes,"
                        + "    `meta` string"
                        + ")\n"
                        + "    WITH (\n"
                        + "    'connector' = 'oracle-cdc-inlong',\n"
                        + "    'job.version' = 'r4',\n"
                        + "    'at-least-once-mode' = 'true',\n"
                        + "    'debezium.log.mining.continuous.mine' = 'true',\n"
                        + "    'debezium.log.mining.strategy' = 'online_catalog',\n"
                        + "    'source.multiple.enable' = 'true',\n"
                        + "    'debezium.snapshot.locking.mode' = 'none',\n"
                        + "    'source.uid' = '7726_1',\n"
                        + "    'debezium.database.tablename.case.insensitive' = 'false',\n"
                        + "    'inlong.table.metadata.columns' = 'opflag|meta.op_type,recordtime|meta.op_ts,receptime|meta.ts',\n"
                        + "    'oceanus.row-kind-string' = 'INS/INS/UPD/UPD',\n"

                        + "%s"
                        + "    'hostname' = '%s',\n"
                        + "    'port' = '%s',\n"
                        + "    'username' = '%s',\n"
                        + "    'password' = '%s',\n"
                        + "    'database-name' = '%s',\n"
                        + "    'schema-name' = '%s',\n"
                        + "    'table-name' = '%s',\n"
                        + "    'scan.startup.mode' = '%s'\n"
                        + ")", optionalParameters, oracleHost, oraclePort, oracleUser, oraclePassword,
                oracleDatabasename, oracleSchemaname, oracleTablesname, startMode);
        LOGGER.info("source is \n {}", sourceSql);
        tableEnv.executeSql(sourceSql);

        return "table_4827_oracle";
    }
}
